package com.gitee.sparta.core.process.feature;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.GlobalThreadPool;
import com.gitee.sparta.annotation.BroadcastListener;
import com.gitee.sparta.annotation.FinalRetry;
import com.gitee.sparta.annotation.SpartaListener;
import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.SpartaErrorEnum;
import com.gitee.sparta.common.exception.Assert;
import com.gitee.sparta.common.helpers.CollHelper;
import com.gitee.sparta.common.helpers.RedisKeyHelper;
import com.gitee.sparta.core.bean.SpartaBeanMapping;
import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.process.SpartaThreadPoolFactory;
import com.gitee.sparta.core.process.operate.MetadataTabOperate;
import com.gitee.sparta.core.process.workers.SpartaListenWorker;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;

/**
 * 任务调度器：
 * 搬运线程与监听线程之间的通讯，当延时任务到达了消费的时机，搬运线程会将 topic 传入到
 * {@see SpartaTaskDispatcherFeature#receive(java.util.List)} 方法内，最终再
 * 调用 {@see SpartaTaskDispatcherFeature#dispatch(java.util.Map)} 方法，将任务提交给监听
 * 线程池，然后搬运线程进入智能睡眠，而提交的任务则是通过线程池运行。
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class SpartaTaskDispatcherFeature<V> {

    /**
     * 队列消息延迟队列，carry 线程应该专注于搬运，过期的消息应该直接放入到延迟队列中
     **/
    public static final LinkedBlockingQueue<Collection<String>> TOPIC_QUEUE = new LinkedBlockingQueue<>();
    /**
     * 广播延迟队列，carry 线程应该专注于搬运，过期的消息应该直接放入到延迟队列中
     **/
    public static final LinkedBlockingQueue<Collection<String>> CHANNEL_QUEUE = new LinkedBlockingQueue<>();
    /**
     * 监听器回调方法缓存
     **/
    private static final Map<String, SpartaBeanMapping> LISTENER_CALLBACK_METHOD_CACHE
            = new ConcurrentHashMap<>(128);
    /**
     * 重试回调方法缓存
     **/
    private static final Map<String, SpartaBeanMapping> RETRY_CALLBACK_METHOD_CACHE
            = new ConcurrentHashMap<>(128);
    /**
     * 广播回调方法缓存
     **/
    private static final Map<String, Set<SpartaBeanMapping>> BROADCAST_CALLBACK_METHOD_CACHE
            = new ConcurrentHashMap<>(128);

    private final SpartaCoreProperties properties;
    private final SpartaListenWorker<V> spartaListenWorker;
    private final MetadataTabOperate<V> metadataTabOperate;
    private final SpartaThreadPoolFactory spartaThreadPoolFactory;

    public SpartaTaskDispatcherFeature(SpartaCoreProperties properties,
                                       SpartaListenWorker<V> spartaListenWorker,
                                       MetadataTabOperate<V> metadataTabOperate,
                                       SpartaThreadPoolFactory spartaThreadPoolFactory) {
        this.properties = properties;
        this.spartaListenWorker = spartaListenWorker;
        this.metadataTabOperate = metadataTabOperate;
        this.spartaThreadPoolFactory = spartaThreadPoolFactory;
        // 启动普通队列的消费 TOPIC_QUEUE
        this.enableQueueConsumer();
        // 启动广播消息的消费 CHANNEL_QUEUE
        this.enableBroadcastConsumer();
    }

    /**
     * 设置 topic 与监听器的绑定关系
     *
     * @param topics     topics: 广播表示 channel，队列表示 topic
     * @param springBean spring bean
     * @param method     可执行方法
     * @param clazz      注解类
     * @param beanName   spring bean 名称
     */
    public static void registerSpartaBeanMapping(String[] topics, Object springBean,
                                                 Method method, Class<?> clazz, String beanName) {
        log.info("[Sparta] Scan sparta bean [{}] mapping method [{}]", beanName, method.getName());

        if (clazz == SpartaListener.class) {
            doRegisterSpartaSingleBeanMapping(LISTENER_CALLBACK_METHOD_CACHE, topics, springBean, method);
        } else if (clazz == FinalRetry.class) {
            doRegisterSpartaSingleBeanMapping(RETRY_CALLBACK_METHOD_CACHE, topics, springBean, method);
        } else if (clazz == BroadcastListener.class) {
            doRegisterSpartaMultiBeanMapping(topics, springBean, method);
        }
    }

    /**
     * 设置 topic 与监听器的绑定关系
     *
     * @param cache      缓存 map 对象
     * @param topics     topics
     * @param springBean spring bean
     * @param method     可执行方法
     */
    private static void doRegisterSpartaSingleBeanMapping(Map<String, SpartaBeanMapping> cache, String[] topics,
                                                          Object springBean, Method method) {
        for (String topic : topics) {
            if (cache.containsKey(topic)) {
                SpartaBeanMapping mapping = cache.get(topic);
                log.error("[Sparta] Topic 已经存在 ：{}, 重复 method :{}, 当前 method： {}",
                        topic, mapping.getCallbackMethod().getName(), method.getName());
                Assert.isTrue(cache.containsKey(topic), SpartaErrorEnum.TOPIC_EXIST);
            }

            SpartaBeanMapping beanMapping = new SpartaBeanMapping(springBean, method);
            cache.put(topic, beanMapping);
        }
    }

    /**
     * 设置 channel 与监听器的绑定关系
     *
     * @param channels   channels
     * @param springBean spring bean
     * @param method     可执行方法
     */
    private static void doRegisterSpartaMultiBeanMapping(String[] channels, Object springBean, Method method) {
        for (String channel : channels) {
            SpartaBeanMapping spartaBeanMapping = new SpartaBeanMapping(springBean, method);
            Set<SpartaBeanMapping> spartaBeanMappings = BROADCAST_CALLBACK_METHOD_CACHE.get(channel);
            if (CollUtil.isEmpty(spartaBeanMappings)) {
                spartaBeanMappings = new HashSet<>(128);
                spartaBeanMappings.add(spartaBeanMapping);
                BROADCAST_CALLBACK_METHOD_CACHE.put(channel, spartaBeanMappings);
            } else {
                spartaBeanMappings.add(spartaBeanMapping);
            }
        }
    }

    /**
     * 获取最终通知的类与方法的映射关系
     *
     * @param topic topic
     * @return {obj, method}
     */
    public static SpartaBeanMapping getFinalRetry(String topic) {
        SpartaBeanMapping beanMapping = RETRY_CALLBACK_METHOD_CACHE.get(topic);
        if (Objects.isNull(beanMapping)) {
            log.warn("[Sparta] Mapping cache, 无法获取到 {} 对应的 finalRetry 方法.", topic);
        }
        return beanMapping;
    }

    /**
     * 遍历所有的 topic, 包装成完整的 readable_tab 的 key
     *
     * @param systemName 系统名称
     * @param isolation  隔离策略
     * @return {systemName}:readable_tab:isolation:topic,{systemName}:readable_tab:isolation:topic1,....
     */
    public static String getReadableTabKeys(String systemName, String isolation) {
        Set<String> serviceTopics = LISTENER_CALLBACK_METHOD_CACHE.keySet();
        if (CollUtil.isEmpty(serviceTopics)) {
            return "";
        }

        StringBuilder readableTabKeys = new StringBuilder();
        int i = 0;
        for (String serviceTopic : serviceTopics) {
            final String readableTabKey = RedisKeyHelper.readableKey(systemName, isolation, serviceTopic);
            readableTabKeys.append(readableTabKey);
            if ((i + 1) != serviceTopics.size()) {
                readableTabKeys.append(SpartaConstant.COMMA);
            }
            ++i;
        }
        return readableTabKeys.toString();
    }

    /**
     * 检查 topic 对应的回调方法是否存在
     *
     * @param topic 被检查的 topic
     * @return true:    存在(放行)  false: 不存在(拒绝)
     * @since v 0.0.2   新增校验监听器，否则无监听器的数据会一直存在 redis 中
     */
    public static boolean topicExistListeners(String topic) {
        return LISTENER_CALLBACK_METHOD_CACHE.containsKey(topic);
    }

    /**
     * 检查 channel 对应的回调方法是否存在
     *
     * @param channel 被检查的 channel
     * @return true:    存在(放行)  false: 不存在(拒绝)
     * @since v 0.0.2   新增校验监听器，否则无监听器的数据会一直存在 redis 中
     */
    public static boolean channelExistListener(String channel) {
        return BROADCAST_CALLBACK_METHOD_CACHE.containsKey(channel);
    }

    /**
     * carry 线程专注于搬运工作，有任务的情况下直接投递到本类中阻塞队列中来，投递完毕后直接返回
     *
     * @param carryTopics fullTopic1,fullTopic2,....
     * @param balance     isolation
     */
    public void deliverDelayQueue(String carryTopics, String balance) {

        Set<String> topics = CollHelper.convertSet(carryTopics);
        if (CollUtil.isEmpty(topics)) {
            return;
        }
        // 发给自己
        if (Objects.equals(balance, ClusterCache.leaderIsolation)) {
            TOPIC_QUEUE.add(topics);
            return;
        }

        // 发给 follower
        this.sendToFollower(balance, topics, ClusterCmd.CONSUME_QUEUE_DATA);
    }

    /**
     * carry 线程专注于搬运工作，有任务的情况下直接投递到本类中阻塞队列中来，投递完毕后直接返回
     *
     * @param carryChannel 过期的广播通道
     * @param balance      负载均衡
     */
    public void deliverBroadcast(String carryChannel, String balance) {
        List<String> channels = CollHelper.convertList(carryChannel, SpartaConstant.COMMA);
        if (CollUtil.isEmpty(channels)) {
            return;
        }

        // 是自己
        if (Objects.equals(balance, ClusterCache.leaderIsolation)) {
            CHANNEL_QUEUE.add(channels);
            return;
        }

        // 是 follower
        this.sendToFollower(balance, channels, ClusterCmd.CONSUME_BROADCAST_DATA);
    }


    /**
     * 发送给 follower
     * 里面包含了广播消息和普通队列消息
     *
     * @param balance 负载
     * @param topics  topics
     * @param command 指令，客户端接收的处理指令
     */
    private void sendToFollower(String balance, Collection<String> topics, byte command) {
        Integer serverId = ClusterCache.getServerId();

        ClusterProtocol pro;
        // netty
        if (ClusterCache.isNetty()) {
            Channel channel = ClusterCache.getChannel(balance);
            Integer sid = ClusterCache.getSid(channel);
            pro = new ClusterProtocol(command, sid, serverId, topics);
            pro.setChannel(channel);
        } else {
            // 发给 follower
            SocketChannel client = ClusterCache.getClient(balance);
            Integer sid = ClusterCache.getSid(client);
            pro = new ClusterProtocol(command, sid, serverId, topics);
            pro.setCli(client);
        }

        AbstractClusterLeader.sendQueue(pro);
    }

    /**
     * {@see deliverDelayQueue() 此方法放消息}
     * 本方法读消息，easy
     * <p>
     * 目的是为了让 carry 线程专注于搬运工作，而非是本类中的 dispatch 工作
     */
    public void enableQueueConsumer() {
        GlobalThreadPool.execute(() -> {
            while (!Thread.interrupted()) {
                Collection<String> topics = null;
                try {
                    topics = TOPIC_QUEUE.take();
                    if (CollUtil.isEmpty(topics)) {
                        continue;
                    }
                } catch (InterruptedException e) {
                    log.error("[Sparta] 从 TOPICS_QUEUE 中 take 数据失败：" + e.getMessage(), e);
                }

                if (CollUtil.isEmpty(topics)) {
                    continue;
                }

                if (this.dispatch(topics)) {
                    TOPIC_QUEUE.clear();
                    Thread.currentThread().interrupt();
                }
            }
        });
    }

    /**
     * {@see deliverBroadcast() 此方法放消息}
     * 本方法读消息，easy
     * <p>
     * 目的是为了让 carry 线程专注于搬运工作，而非是本类中的 dispatch 工作
     */
    public void enableBroadcastConsumer() {
        if (!this.properties.isEnableBroadcast()) {
            return;
        }
        GlobalThreadPool.execute(() -> {
            while (!Thread.interrupted()) {
                Collection<String> channels = null;
                try {
                    channels = CHANNEL_QUEUE.take();
                } catch (InterruptedException e) {
                    log.error("[Sparta] 从 CHANNEL_QUEUE 中 take 数据失败：" + e.getMessage());
                    e.printStackTrace();
                }
                if (CollUtil.isEmpty(channels)) {
                    continue;
                }

                for (String channel : channels) {
                    // 直接将内容读取出来
                    Metadata<V> metadata = this.metadataTabOperate.readDelMetadata(channel);

                    Set<SpartaBeanMapping> spartaBeanMappings = BROADCAST_CALLBACK_METHOD_CACHE
                            .get(metadata.getTopic());
                    if (CollUtil.isNotEmpty(spartaBeanMappings)) {
                        for (SpartaBeanMapping mapping : spartaBeanMappings) {
                            spartaThreadPoolFactory.getBroadcastThreadPool().execute(() -> {
                                this.spartaListenWorker.broadcast(mapping, metadata);
                            });
                        }
                    }
                }
            }
        });
    }

    /**
     * 分配任务，监听线程拿到待消费的 TOPIC，直接分配到线程池内
     *
     * @param topicSet 监听的 topic, @SpartaListener 中的 topics 属性
     */
    private boolean dispatch(Collection<String> topicSet) {
        if (this.spartaThreadPoolFactory.getListenersThreadPool().isShutdown()) {
            return true;
        }

        try {
            for (final String topic : topicSet) {
                this.spartaThreadPoolFactory.getListenersThreadPool().execute(() -> {
                    SpartaBeanMapping mapping = LISTENER_CALLBACK_METHOD_CACHE.get(topic);
                    this.spartaListenWorker.doWorker(topic, mapping);
                });
            }
            return false;
        } catch (RejectedExecutionException ex) {
            return true;
        }
    }
}
